package com.yangzb;

import cn.hutool.core.io.FileUtil;
import com.google.common.base.Stopwatch;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Callable;

/**
 *
 * @author: yangzb
 * @date 2023/1/3 2:28 PM
 **/
@Slf4j
public class ClickhouseWriter implements Callable<TaskInfo> {
    protected ClickHouseConfig clickHouseConfig;
    protected TaskInfo taskInfo;

    public ClickhouseWriter(TaskInfo taskInfo, ClickHouseConfig clickHouseConfig){
        this.clickHouseConfig = clickHouseConfig;
        this.taskInfo = taskInfo;
    }

    @Override
    public TaskInfo call(){
        Stopwatch stopwatch  = Stopwatch.createStarted();
        try {
            ClickHouseUtil.importFile(clickHouseConfig.getUris(),taskInfo.getLocalFile(),clickHouseConfig.getClickhouseTableName(),clickHouseConfig.getFormat(),clickHouseConfig.getClickHouseCompression());
        } catch (Exception e) {
            log.error("",e);
            taskInfo.setException(e);
            taskInfo.setFailedStep(TaskInfo.FailedStep.IMPORT);
        }finally {
            try {
                FileUtil.del(taskInfo.getLocalFile());
            } catch (Exception e) {
                log.error("",e);
            }
            stopwatch.stop();
            log.info("导入:{},耗时：{}",taskInfo.getHdfsFile(),stopwatch);
        }
        return taskInfo;
    }
}
